package dyyx.demo;
import javax.annotation.Nullable;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class WatermarkDemo2 {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<String> dataStream = env
                .socketTextStream("127.0.0.1", 9002).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {
                    long currentTimeStamp = 0l;
                    long maxDelayAllowed = 0l;
                    long currentWaterMark;

                    @Nullable
                    @Override
                    public Watermark getCurrentWatermark() {
                        currentWaterMark = currentTimeStamp - maxDelayAllowed;
                        return new Watermark(currentWaterMark);
                    }

                    @Override
                    public long extractTimestamp(String s, long l) {
                        String[] arr = s.split(",");
                        long timeStamp = Long.parseLong(arr[1]);
                        currentTimeStamp = Math.max(timeStamp, currentTimeStamp);
                        System.out.println("Key:" + arr[0] + ",EventTime:" + timeStamp + ",waterMark:" + currentWaterMark);
                        return timeStamp;
                    }
                });

        dataStream.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String s) throws Exception {
            	String[]arr = s.split(",");
                return new Tuple2<String, String>(arr[0], arr[1]);
            }
        }).keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //.apply(new MyWinFunction())
                
                ;
        // TODO reduce aggregate apply

        

        env.execute("WatermarkDemo2");
		
	}
	
	private static class MyWinFunction implements  WindowFunction<Tuple2<String, String>, String, String, TimeWindow>{
		public void apply(String key, TimeWindow window, Iterable<Tuple2<String, String>> input, Collector<String> out) throws Exception{
			for(Tuple2<String, String> item:input) {
				String s = key+" "+ item.f0+" "+item.f1+" "+window;
				out.collect(s);
			}
			
		}
	}
}
